package com.alpaca.recommend.bean;

import com.alpaca.recommend.hbase.HTableClients;
import com.alpaca.recommend.hbase.activerecord.HModel;
import com.alpaca.recommend.util.MD5Util;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.io.UnsupportedEncodingException;

/**
 * Created by justin on 15-2-10.
 */
public class TagItem extends HModel<TagItem> {

    public static final TagItem dao = new TagItem();

    public static final String TABLE_NAME = "tag_items";

    public static final byte[] TAG_ITEM_FAMILY = Bytes.toBytes("t");

    public static final byte[] TAG_ITEM_RANK_COLUMN = Bytes.toBytes("r");

    public static final byte[] TAG_ITEM_COUNT_COLUMN = Bytes.toBytes("c");

    public static Put getPut(String channelCode, String itemId, String TAG, int rank, int count) throws UnsupportedEncodingException {
        String rowKey = MD5Util.md5(String.valueOf(channelCode)) + MD5Util.md5(String.valueOf(TAG)) + itemId;
        System.out.println("rowKey: " + rowKey);
        Put put = new Put(Bytes.toBytes(rowKey));
        put.add(TAG_ITEM_FAMILY, TAG_ITEM_RANK_COLUMN, Bytes.toBytes(rank));
        put.add(TAG_ITEM_FAMILY, TAG_ITEM_COUNT_COLUMN, Bytes.toBytes(count));
        return put;
    }


    public static void main(String[] args) throws IOException {
        Table tagItemsTable = new HTableClients().getHTable(TagItem.TABLE_NAME);
        String uTagsRowKey = "a0b923820dcc509a7a99787bd197eacd司马懿";
        String channelcode = uTagsRowKey.substring(0, 16);
        String userId = uTagsRowKey.substring(16, 32);
        String tag = uTagsRowKey.substring(32);
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(channelcode + MD5Util.md5(tag)));
        scan.setStopRow(Bytes.toBytes(channelcode + MD5Util.md5(tag) + 'g'));
        ResultScanner rs = tagItemsTable.getScanner(scan);
        for (Result result : rs) {
            Cell cell = result.getColumnLatestCell(TagItem.TAG_ITEM_FAMILY, TagItem.TAG_ITEM_RANK_COLUMN);
            System.out.println(Bytes.toInt(CellUtil.cloneValue(cell)));
        }
    }

    /**
     * 初始化方法.
     *
     * @param channelCode 频道号.
     * @param itemId      项目id
     * @param tag         标签.
     */
    public void init(String channelCode, String itemId, String tag) {
        //设置tag.
        String rowKey = MD5Util.md5(String.valueOf(channelCode)) + MD5Util.md5(String.valueOf(tag)) + itemId;
        System.out.println("rowKey: " + rowKey);
        setRow(rowKey);
        //设置rank
        set("t", "r", 1);
        //设置count
        set("t", "c", 1);
    }

    /**
     * 获取rowkey.
     *
     * @param channelCode 频道号
     * @param itemId      项目id
     * @param tag         标签
     * @return
     */
    public static String getRowkey(String channelCode, String itemId, String tag) {
        return MD5Util.md5(String.valueOf(channelCode)) + MD5Util.md5(String.valueOf(tag)) + itemId;
    }
}